#pragma warning(disable : 4355)
#include "Avalon/Network/TcpServerSocket.hpp"
#include <boost/bind.hpp>
#include <boost/lexical_cast.hpp>
#include "Avalon/IO/NotConnectedException.hpp"
#include "Avalon/Network/SocketException.hpp"
#include "Avalon/Network/SocketThreadPool.hpp"
#include "Avalon/Network/SocketType.hpp"
#include "Avalon/Network/TcpSocketConnection.hpp"
#include "Avalon/Network/TcpSocketReader.hpp"
#include "Avalon/Network/TcpSocketWriter.hpp"

using namespace Avalon;
using namespace Avalon::IO;
using namespace Avalon::Network;
using namespace Avalon::Network::Detail;
using namespace Avalon::Threading;
using namespace boost;
using namespace boost::asio;
using namespace boost::posix_time;
using namespace std;
using boost::asio::ip::tcp;

TcpServerSocket::TcpServerSocket(const IpAddress& address)
    : m_address(address),
      m_isOpen(false),
      m_ioService(SocketThreadPool::GetInstance().GetService()),
      m_pendingClose(false),
      m_acceptSlot(bind(&TcpServerSocket::OnAccept, this, _1)) {}

TcpServerSocket::~TcpServerSocket() {
  Close();
}

void TcpServerSocket::Open() {
  lock_guard<mutex> lock(m_mutex);
  if(m_isOpen) {
    return;
  }
  tcp::resolver resolver(m_ioService);
  tcp::resolver::query query(m_address.first,
    lexical_cast<string>(m_address.second));
  system::error_code error;
  tcp::resolver::iterator endpointIterator = resolver.resolve(query, error);
  if(error != 0) {
    BOOST_THROW_EXCEPTION(SocketException(error.value(), error.message()));
  }
  m_acceptor = new tcp::acceptor(m_ioService, *endpointIterator);
  m_isOpen = true;
}

void TcpServerSocket::Accept(Out<Async<Channel*> > channel) {
  unique_lock<mutex> lock(m_mutex);
  if(!m_isOpen) {
    lock.unlock();
    channel->SetException(NotConnectedException());
    return;
  }
  m_acceptResult = channel.Get();
  tcp::acceptor& r_acceptor = *static_cast<tcp::acceptor*>(m_acceptor);
  m_socket.reset(new SocketType(m_ioService));
  r_acceptor.async_accept(m_socket->m_socket, m_acceptSlot);
}

bool TcpServerSocket::IsOpen() const {
  lock_guard<mutex> lock(m_mutex);
  return m_isOpen;
}

void TcpServerSocket::Close() {
  shared_ptr<ClosedSlot> closedSlot;
  {
    unique_lock<mutex> lock(m_mutex);
    if(!m_isOpen) {
      return;
    }
    tcp::acceptor& r_acceptor = *static_cast<tcp::acceptor*>(m_acceptor);
    r_acceptor.close();
    m_pendingClose = true;
    while(m_acceptResult != NULL) {
      m_closedCondition.wait(lock);
    }
    m_pendingClose = false;
    delete static_cast<tcp::acceptor*>(m_acceptor);
    m_isOpen = false;
    closedSlot = m_closedSlot;
  }
  (*closedSlot)();
}

void TcpServerSocket::SetClosedSlot(const ClosedSlot& slot, Out<bool> isOpen) {
  lock_guard<mutex> lock(m_mutex);
  *isOpen = m_isOpen;
  m_closedSlot.reset(new ClosedSlot(slot));
}

void TcpServerSocket::OnAccept(const system::error_code& error) {
  unique_lock<mutex> lock(m_mutex);
  if(error != 0) {
    Async<Channel*>* acceptResult = m_acceptResult;
    m_acceptResult = NULL;
    if(m_pendingClose) {
      m_closedCondition.notify_one();
    }
    lock.unlock();
    acceptResult->SetException(SocketException(error.value(), error.message()));
    return;
  }
  shared_ptr<TcpSocketReader> reader(new TcpSocketReader(m_socket));
  shared_ptr<TcpSocketWriter> writer(new TcpSocketWriter(m_socket));
  shared_ptr<TcpSocketConnection> connection(new TcpSocketConnection(
    true, IpAddress(m_socket->m_socket.remote_endpoint().address().to_string(),
    m_socket->m_socket.remote_endpoint().port()), m_socket, reader, writer));
  TcpClientSocketChannel* channel = new TcpClientSocketChannel(connection,
    reader, writer);
  m_socket.reset();
  Async<Channel*>* acceptResult = m_acceptResult;
  m_acceptResult = NULL;
  if(m_pendingClose) {
    m_closedCondition.notify_one();
  }
  lock.unlock();
  acceptResult->SetResult(channel);
}
